package tech.abing.spider.task.pipeline;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.LinkedHashMap;
import java.util.List;

import org.apache.log4j.Logger;

import tech.abing.spider.component.Pipeline;
import tech.abing.spider.constants.FinalConstants;
import tech.abing.spider.model.Proccessable;
import tech.abing.spider.task.model.CloseEntity;
import tech.abing.spider.task.model.PayLogClose;
import tech.abing.spider.task.model.PayRecord;
import tech.abing.spider.task.model.TaskTrade;
import tech.abing.spider.util.DateUtils;
import tech.abing.spider.util.ProccessableUtil;
import tech.abing.spider.util.RedisUtil;
import tech.abing.spider.util.StringUtils;
import tech.abing.spider.util.SysUtil;

public class PayLogPipeline implements Pipeline {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private static Logger logger = Logger.getLogger(PayLogPipeline.class);

	private static FileWriter writer = null;
	
	private static String fileName = "", path = "";
	
	private static LinkedHashMap<TaskTrade,Integer> taskMap = null;//重新入队队尾的任务
	
	private static RedisUtil redisUtil = null;
	
	public PayLogPipeline() {
		
		if(redisUtil == null){
			redisUtil = RedisUtil.getInstance();
			logger.info("Redis连接池创建成功！");
		}
		
		if(taskMap == null){
			taskMap = new LinkedHashMap<TaskTrade,Integer>();
		}
		
		fileName = "pay_"+DateUtils.getDate()+"_"+SysUtil.getUUID().replace("-", "")+".dat";
		logger.info("=========>输出交易记录的文件名："+fileName);
		path = FinalConstants.CRAWL_STORAGE_FOLDER_PREFIX+"TMALL/plg/"+DateUtils.getDate()+"/";
		
		File file = new File(path);
		if(!file.exists()){
			file.mkdirs();
		}
		file = new File(path+fileName);
		
		// 打开一个写文件器，构造函数中的第二个参数true表示以追加形式写文件
		try {
			if(writer == null){
				writer = new FileWriter(file, true);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	

	@SuppressWarnings("unchecked")
	@Override
	public void proccessData(List<Proccessable> objectContainer) {
		
		if (ProccessableUtil.instanceOfClass(objectContainer, PayRecord.class)) {// 判断集合类型是否是PayRecord
			List<PayRecord> list = ProccessableUtil.convert(objectContainer, PayRecord.class);// 转换
			try {
				String split = FinalConstants.SPLIT;
				StringBuffer content = new StringBuffer();
				
				for(PayRecord pr : list){
					
					//去重任务,并记录抓取交易记录的次数
					TaskTrade taskTrade = new TaskTrade(pr.getPrdNo(), pr.getShopId(), pr.getType(), pr.getMaxPayDate()+" "+pr.getMaxPayTime(), pr.getCatchDate(), pr.getMonthSale());
					if(!taskMap.containsKey(taskTrade)){
						taskMap.put(taskTrade, 1);
					}else{
						taskMap.put(taskTrade, taskMap.get(taskTrade) + 1);
					}
					content.append(pr.getShopId()).append(split).append(pr.getPrdNo()).append(
							split).append(pr.getSku()).append(split+pr.getNum()).append(
							split).append(pr.getPayPrice()).append(split).append(pr.getNickName())
							.append(split).append(pr.getRank()).append(split).append(pr.getLevel()).append(
							split).append(pr.getPayDate()).append(split).append(pr.getPayTime())
							.append(split).append(pr.getCatchDate()).append(FinalConstants.ENTER_CHAR);
				}
				writer.write(content.toString());
				writer.flush();
			} catch (IOException e) {
				logger.error("=========>离线写交易数据失败："+fileName);
			}
		}
		
		//重新入队队尾
		LinkedHashMap<TaskTrade, Integer> taskMapClone = (LinkedHashMap<TaskTrade, Integer>) taskMap.clone();
		
		if(taskMapClone.size() > 1){//包含两个就说明前一个产品的交易记录抓取完成，可以重新入队队尾
			
			int count = 0;
			for(TaskTrade tt : taskMapClone.keySet()){
				pushRedis(tt);
				
				taskMap.remove(tt);
				
				if(++count == taskMapClone.size() - 1){//留最后一个不入队，等到全部抓取完成后入队
					break;
				}
				
			}
			
		}
	}

	@Override
	public void destory() {
		
		//剩下的入队队尾
		if(taskMap.size() > 0){
			
			for(TaskTrade tt : taskMap.keySet()){
				
				pushRedis(tt);
				
			}
			taskMap.clear();
		}
		if(redisUtil != null){
			redisUtil.destroy();
			logger.info("Redis连接池关闭成功！");
		}
		
		logger.info("=========>清理IO资源");
		if(writer != null){
			try {
				writer.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
	}


	private void pushRedis(TaskTrade tt) {
		int tradeCount = taskMap.get(tt);
		
		String unsaleTask = SysUtil.getValue("redis.task_trade_product_id_unsale");
		String normalTask = SysUtil.getValue("redis.task_trade_product_id_normal");
		String hotTask = SysUtil.getValue("redis.task_trade_product_id_hot");
		
		String productSuccess = SysUtil.getValue("redis.trade_product_id_success");
		
		redisUtil.sadd(tt.getProdId(), productSuccess);
		
		if(tradeCount >= FinalConstants.PRODUCT_HOT_SALE_MIN_DAY){//热销
			
			//push到制定队列中
			try {
				redisUtil.lpush(tt, hotTask);
				logger.info("产品ID["+tt+"]入队队尾["+hotTask+"]成功");
			} catch (Exception e) {
				e.printStackTrace();
				logger.error("产品ID["+tt+"]入队队尾["+hotTask+"]失败", e);
			}
			
		}else if(tradeCount >= FinalConstants.PRODUCT_NORMAL_SALE_MIN_DAY && tradeCount < FinalConstants.PRODUCT_HOT_SALE_MIN_DAY){//普通
			
			//push到制定队列中
			try {
				redisUtil.lpush(tt, normalTask);
				logger.info("产品ID["+tt+"]入队队尾["+normalTask+"]成功");
			} catch (Exception e) {
				e.printStackTrace();
				logger.error("产品ID["+tt+"]入队队尾["+normalTask+"]失败", e);
			}
			
		}else if(tradeCount > FinalConstants.PRODUCT_UNSALE_SALE_MIN_DAY && tradeCount < FinalConstants.PRODUCT_NORMAL_SALE_MIN_DAY){//滞销
			
			//push到制定队列中
			try {
				redisUtil.lpush(tt, unsaleTask);
				logger.info("产品ID["+tt+"]入队队尾["+unsaleTask+"]成功");
			} catch (Exception e) {
				e.printStackTrace();
				logger.error("产品ID["+tt+"]入队队尾["+unsaleTask+"]失败", e);
			}
			
		}
	}

}
